package m2;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.List;
import java.util.Scanner;

public class Producer {

        static String[] msgs = {
                "15103111039,创建",
                "15103111065,创建",
                "15103111039,付款",
                "15103117235,创建",
                "15103111065,付款",
                "15103117235,付款",
                "15103111065,完成",
                "15103111039,推送",
                "15103117235,完成",
                "15103111039,完成"
        };

    public static void main(String[] args) throws Exception {
        DefaultMQProducer p = new DefaultMQProducer("producer");
        p.setNamesrvAddr("192.168.64.141:9876;192.168.64.141:9876");
        p.start();


        String topic = "Topic";
        String tag = "TagA";
        for (String s : msgs) {
            System.out.println("按回车发送此消息: " + s);
            new Scanner(System.in).nextLine();
              //创建消息对象和传参
            Message msg = new Message(topic, tag, s.getBytes());
            String[] a = s.split(",");//分割消息为字符串数组
            long orderId = Long.parseLong(a[0]);//转型
              //发送消息，带参
            SendResult r = p.send(msg, new MessageQueueSelector() { //消息队列选择器处理消息
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    Long orderId = (Long) o;
                    //订单id对【队列数量】取余, 相同订单id得到相同的队列索引
                    long index = orderId % list.size();

                    System.out.println("消息已发送到: " + list.get((int) index));
                    return list.get((int) index);

                }
            }, orderId);
            System.out.println(r+"\n\n");
        }
    }
}
